package com.aiwiown.snackmq.broker.handler.console;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.common.auth.Action;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.protocol.TopicPartition;
import com.aiwiown.snackmq.storage.StorageService;
import com.aiwiown.snackmq.storage.exception.PartitionNotFoundException;
import com.aiwiown.snackmq.storage.service.PartitionedStorageService;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

/**
 * 处理分区修复请求的处理器。
 * 该处理器负责修复指定分区中损坏的段文件。
 */
@Slf4j
public class RepairPartitionHandler implements MessageHandler {

    private final StorageService storageService;
    private final AuthorizationService authorizationService;

    public RepairPartitionHandler(StorageService storageService, AuthorizationService authorizationService) {
        this.storageService = storageService;
        this.authorizationService = authorizationService;
    }

    @Override
    public void handle(RequestContext ctx, Message request) {
        try {
            String topic = request.getTopic();
            int partitionId = Integer.parseInt(request.getProperty("partition"));

            // 授权检查
            authorizationService.authorize(
                    request.getProperty("authToken"),
                    Action.CONSUME,
                    new AuthorizationService.Resource("TOPIC", topic)
            );

            // 执行修复
            repairPartition(topic, partitionId);

            Message response = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.SUCCESS)
                    .build();
            ctx.writeResponse(response);

        } catch (PartitionNotFoundException e) {
            log.warn("Partition not found for repair: {}", request, e);
            Message errorResponse = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.FAILED)
                    .errorMessage("Partition not found: " + e.getMessage())
                    .build();
            ctx.writeResponse(errorResponse);
        } catch (Exception e) {
            log.error("Failed to repair partition for request: {}", request, e);
            Message errorResponse = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.FAILED)
                    .errorMessage("Repair failed: " + e.getMessage())
                    .build();
            ctx.writeResponse(errorResponse);
        }
    }

    private void repairPartition(String topic, int partitionId) throws Exception {
        if (!(storageService instanceof PartitionedStorageService)) {
            throw new UnsupportedOperationException("Repair only supported for PartitionedStorageService");
        }

        PartitionedStorageService partitionedStorage = (PartitionedStorageService) storageService;
        TopicPartition topicPartition = new TopicPartition(topic, partitionId);
        
        log.info("开始修复分区 {}", topicPartition);
        
        // 获取损坏的段列表
        List<Long> corruptedSegments = partitionedStorage.getCorruptedSegments(topicPartition);
        if (corruptedSegments.isEmpty()) {
            log.info("分区 {} 没有发现损坏的段", topicPartition);
            return;
        }
        
        log.info("分区 {} 发现 {} 个损坏的段: {}", topicPartition, corruptedSegments.size(), corruptedSegments);
        
        // 执行修复
        partitionedStorage.repairPartition(topicPartition);
        
        log.info("分区 {} 修复完成", topicPartition);
    }

    @Override
    public MessageType getMessageType() {
        return MessageType.REPAIR_PARTITION_REQUEST;
    }
} 